package com.ibm.cps.processors.vector;


import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.log4j.Logger;

import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.processors.Info;
import com.ibm.cps.processors.SchemaInheritiveProcessor;
import com.ibm.cps.sql.expression.ExpressionNode;
import com.ibm.cps.sql.expression.FieldEvaluator;
import com.ibm.cps.sql.expression.ValueExpression;
import com.ibm.cps.sql.expression.function.meta.FieldNameParam;
import com.ibm.cps.sql.expression.function.meta.IgnoredInFunctionCall;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.ErrorCode;
import com.ibm.util.exception.CPSException;

/*		
    {
        "processorid": "562729fdda0a6ddb89ae8dfc",
        "parentprocessors": "5627347916d3b38d847a9b18",
        "groupid": "{{groupid}}",
        "type": "vector calculation",
        "name": "vector-calculation",
        "leftOperand":{"type":"field","name":"a_array"},
		"rightOperand":{"type":"field","name":"b_array"},
		"outputFieldName":"result",
		"operator":"+"
    },
 * checked.
 */
@Info(Name=MessageFields.VECTOR_CALCULATION,Type=Info.ProcessorType.Other)
public class VectorCalculationProcessor extends SchemaInheritiveProcessor{
	private static final Logger logger = Logger.getLogger(VectorCalculationProcessor.class);
	@ValueParam @IgnoredInFunctionCall
	private String operatorString;
	@ValueParam @IgnoredInFunctionCall
	private String leftOperandClass;
	@ValueParam @IgnoredInFunctionCall
	private String leftOperandString;
	@ValueParam	@IgnoredInFunctionCall
	private String rightOperandClass;
	@ValueParam @IgnoredInFunctionCall
	private String rightOperandString;
	@FieldNameParam @IgnoredInFunctionCall
	private String outputFieldName;
	
	private ExpressionNode leftOperand;
	private ExpressionNode rightOperand;
	private OperatorFunction operator;
	public static final String OPERATOR = "operator";
	public static final String LEFT_OPERAND = "leftOperand";
	public static final String RIGHT_OPERAND = "rightOperand";
	public static final String OUTPUT_FIELD_NAME = "outputFieldName";

	public VectorCalculationProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
	}
	

	
	private ExpressionNode createOperand(String operandJson, String operandClassName) throws CPSException { 
		try {
			ObjectMapper mapper = new ObjectMapper();
			Class<?> leftOperandType = Class.forName(operandClassName);
			ExpressionNode node = (ExpressionNode)mapper.readValue(operandJson, leftOperandType);
			return node;
		} catch (ClassNotFoundException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "expression class not found: " + operandClassName, e);
		} catch (JsonParseException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "failed to parse expression json: " + operandJson + " to " + operandClassName, e);
		} catch (JsonMappingException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "failed to parse expression json: " + operandJson + " to " + operandClassName, e);
		} catch (IOException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "failed to parse expression json: " + operandJson + " to " + operandClassName, e);
		}
	}
	
	private Double[] calculateOutputVector(Double[] leftValues, Double[] rightValues) throws CPSException{
		logger.info(this.getProcessorId() + ": leftOperatorDimension = " + leftValues.length + ", rightOperatorDimension = " + rightValues.length);
		// scalar calculation
		if(leftValues.length == 1 && rightValues.length == 1){
			logger.info(this.getProcessorId() + ": performing scalar calculation");
			return new Double[]{operator.evaluate(leftValues[0], rightValues[0])};
		}else if(leftValues.length == 1){ // num
			logger.info(this.getProcessorId() + ": performing scalar value versus vector calculation");
			Double[] output = new Double[rightValues.length];
			double leftValue = leftValues[0];
			for (int i = 0; i < rightValues.length; i++) {
				output[i] = operator.evaluate(leftValue, rightValues[i]);
			}
			return output;
		}else if(rightValues.length == 1){
			logger.info(this.getProcessorId() + ": performing scalar value versus vector calculation");
			Double[] output = new Double[leftValues.length];
			double rightValue = rightValues[0];
			for (int i = 0; i < leftValues.length; i++) {
				output[i] = operator.evaluate(leftValues[i], rightValue);
			}
			return output;
		}else{
			logger.info(this.getProcessorId() + ": performing vector to vector calculation");
			if(leftValues.length != rightValues.length){
				throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "Dimension mismatch in Vector Calculation Processor, left operand is " + leftValues.length + " dimension, right operand is " + rightValues.length + " dimension");
			}
			Double[] output = new Double[leftValues.length];
			for (int i = 0; i < leftValues.length; i++) {
				output[i] = operator.evaluate(leftValues[i], rightValues[i]);
			}
			return output;
		}
	}

	@Override
	public void dispose() {
	}


	@Override
	protected List<NewField> defineFieldsToBeAdded() {
		List<NewField> fields = new ArrayList<>();
		fields.add(new NewField(outputFieldName,"double[]"));
		return fields;
	}

	@Override
	protected Collection<ParsedDataMessage> runProcessorLogic(ITimeSeriesSchema parentSchema,
			ITimeSeriesSchema outputSchema, ParsedDataMessage message) throws CPSException {
		Collection<ParsedDataMessage> messages = new ArrayList<>();
		ParsedDataMessageBuilder outputBuilder = ParsedDataMessageBuilder.createInhertiveBuilder(outputSchema, message, parentSchema);
		
		Double[] leftVector = SchemaUtility.extractValues(parentSchema, message, leftOperand);
		Double[] rightVector = SchemaUtility.extractValues(parentSchema, message, rightOperand);
		
		int leftDimension = leftVector.length;
        int rightDimension = rightVector.length;
        if(leftDimension != rightDimension && leftDimension != 1 && rightDimension != 1){
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "Dimension mismatch in Vector Calculation Processor, left operand is " + leftDimension + " dimension, right operand is " + rightDimension + " dimension");
		}
        Double[] outputVector = calculateOutputVector(leftVector, rightVector);
		
        logger.info("==============VectorCalculation input============");
		logger.info("operator: " + operator);
        logger.info("leftValues: " + Arrays.toString(leftVector));
        logger.info("rightValues: " + Arrays.toString(rightVector));
       
        outputBuilder.append(outputFieldName, outputVector);
		logger.info("==============VectorCalculation output============");
		logger.info(Arrays.toString(this.getTimeSeriesSchema().getTagNames()));
		logger.info(Arrays.toString(outputVector));
		
		messages.add(outputBuilder.toParsedDataMessage());
		return messages;
	}

	@Override
	public void initialize() throws CPSException {
		leftOperand = createOperand(leftOperandString, leftOperandClass);
		rightOperand = createOperand(rightOperandString, rightOperandClass);
		operator = OperatorFunctions.getOperatorFunction(operatorString);
	}

}
